package indi.mozping.lock;

import org.I0Itec.zkclient.IZkDataListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @author by mozping
 * @Classname ZookeeperDistributeListenLock
 * @Description zookeeper基于相互监听的分布式锁
 * 实现思路是：
 * * 1.客户端在zookeeper的指定路径下创建临时节点有序的节点，节点会按照序号排序
 * * 2.客户端创建临时有序节点成功后会判断自身是不是第一个节点，如果是，那么就获得了锁
 * * 3.如果自己不是，那么就找到自己前面的一个节点，并监听它，一旦前面的节点被删除，自己就可以尝试去获得锁了，没有被删除的话，自身会阻塞
 * * 4.可参考pic下面的流程图
 * @Date 2019/4/19 14:44
 */
public class ZookeeperDistributeListenLock extends ZookeeperAbstractLock {

    private static final Logger LOG = LoggerFactory.getLogger(ZookeeperDistributeListenLock.class);

    private String beforePath;
    private String currentPath;
    private CountDownLatch countDownLatch = null;

    public ZookeeperDistributeListenLock() {
        if (!zkClient.exists(PATH2)) {
            zkClient.createPersistent(PATH2);
        }
    }

    @Override
    boolean tryLock() {
        //如果currentPath为空，则尝试加锁，第一次加锁赋值currentPath,并创建临时有序节点
        if (currentPath == null || currentPath.length() < 1) {
            currentPath = zkClient.createEphemeralSequential(PATH2 + '/', "lock");
        }
        //获取所有临时节点并且排序
        List<String> childrens = zkClient.getChildren(PATH2);
        Collections.sort(childrens);
        //如果自己是第一个节点，那么就认为自己获取到锁了
        if (currentPath.equals(PATH2 + '/' + childrens.get(0))) {
            return true;
        } else {
            //如果当前节点不是排名第一，那么就获取其前面一个节点，并在后面监听前面的节点
            int wz = Collections.binarySearch(childrens, currentPath.substring(7));
            beforePath = PATH2 + '/' + childrens.get(wz - 1);
        }
        return false;
    }

    @Override
    void waitLock() {
        IZkDataListener listener = new IZkDataListener() {
            public void handleDataChange(String dataPath, Object data) throws Exception {

            }

            public void handleDataDeleted(String dataPath) throws Exception {
                if (countDownLatch != null) {
                    countDownLatch.countDown();
                }
            }
        };
        //监听前面一个节点是否被删除，前面节点一旦被删除，自己就有机会获取锁了
        zkClient.subscribeDataChanges(beforePath, listener);
        if (zkClient.exists(beforePath)) {
            countDownLatch = new CountDownLatch(1);
            try {
                //如果起那么的节点一直存在，自己就等着，直到前面的节点被删除，方法再返回，
                // 这里的await就相当于一直阻塞在这里等着，删除的动作在listen里面实现
                countDownLatch.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        //删除监听
        zkClient.unsubscribeDataChanges(beforePath, listener);
    }

    public void unLock() {
        if (zkClient != null) {
            zkClient.delete(currentPath);
            zkClient.close();
            LOG.info("释放锁资源...");
        }
    }
}
